package com.mytest.file;

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.zip.ZipEntry;
import java.util.zip.ZipInputStream;

@Service
@Slf4j
public class FileProcessingService {

//    private final DatabaseService databaseService;
//
//    @Autowired
//    public FileProcessingService(DatabaseService databaseService) {
//        this.databaseService = databaseService;
//    }

    public Mono<Void> processFile(Path zipFilePath) {
        // 异步解压文件
        return Mono.fromRunnable(() -> unzipFile(zipFilePath))
                .thenMany(Flux.fromIterable(getExtractedFilePaths(zipFilePath)))
//                .flatMap(this::processFileContent)  // 处理解压后的文件
                .then(); // 处理完成
    }

    private void unzipFile(Path zipFilePath) {
        // 解压文件
        try (ZipInputStream zipInputStream = new ZipInputStream(Files.newInputStream(zipFilePath))) {
            ZipEntry entry;
            while ((entry = zipInputStream.getNextEntry()) != null) {
                // 解压文件到临时目录
                Path extractedFilePath = Paths.get("extracted", entry.getName());
                Files.createDirectories(extractedFilePath.getParent());
                try (FileOutputStream outputStream = new FileOutputStream(extractedFilePath.toFile())) {
                    byte[] buffer = new byte[4096];
                    int bytesRead;
                    while ((bytesRead = zipInputStream.read(buffer)) != -1) {
                        outputStream.write(buffer, 0, bytesRead);
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("Failed to unzip file", e);
        }
    }

    private List<Path> getExtractedFilePaths(Path zipFilePath) {
        // 返回解压后的文件路径列表
        File extractedDir = new File("extracted");
        return Arrays.stream(extractedDir.listFiles())
                .map(File::toPath)
                .collect(Collectors.toList());
    }

//    private Mono<Void> processFileContent1(Path filePath) {
//        return Mono.fromCallable(() -> Files.newBufferedReader(filePath))
//                .flatMapMany(reader -> Flux.generate(() -> reader, (r, sink) -> {
//                    try {
//                        String line = r.readLine();
//                        if (line != null) {
//                            sink.next(line);  // 将每一行数据传递到流
//                        } else {
//                            r.close();  // 关闭资源
//                            sink.complete();  // 文件读取完毕
//                        }
//                    } catch (IOException e) {
//                        sink.error(e);  // 捕获异常
//                    }
//                }))
//                .buffer(1000)  // 每 1000 行数据分成一组
//                .doOnNext(lines -> log.info("Processing batch of size {}: {}", lines.size(), lines))
////                .flatMap(lines ->
////                        databaseService.saveBatch(lines)) // 批量保存
//                .then();
//    }


    private static Mono<Void> processFileContent(Path filePath) {
        AtomicInteger i = new AtomicInteger(1);
        return Mono.fromCallable(() -> Files.newBufferedReader(filePath))
                .flatMapMany(reader -> Flux.generate(
                        () -> reader,  // 提供初始状态
                        (r, sink) -> {
                            try {

                                String line = r.readLine();
                                if (line != null) {
                                    log.info("line   **************************************: {}", i.getAndIncrement());
                                    sink.next(line); // 传递行数据到流
                                } else {
                                    sink.complete(); // 文件读取完毕
                                    r.close(); // 关闭资源
                                }
                            } catch (IOException e) {
                                sink.error(e); // 捕获异常
                                try {
                                    r.close(); // 出现异常时也要关闭资源
                                } catch (IOException ignored) {
                                    // 忽略二次关闭异常
                                }
                            }
                            return r; // 返回当前状态（即 reader）
                        }))
                .buffer(1000) // 每 1000 行数据分组
                .doOnNext(lines -> log.info("Processing batch of size {}", lines.size()))
//                        log.info("Processing batch of size {}: {}", lines.size(), lines))
                //                .flatMap(lines ->
//                        databaseService.saveBatch(lines)) // 批量保存
                .then();
    }


    public static void main(String[] args) {
        // 准备测试文件路径
        Path testFilePath = Paths.get("C:\\资料\\行业\\环保\\噪声\\铁塔计算示范区计算数据20241117\\tower20\\tower20.ssrec");
        log.info("start");
        processFileContent(testFilePath).block();
        log.info("end");
    }

}
